package com.mq.redis.queue.bean;

import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisConnectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.util.CollectionUtils;

import com.mq.model.AckPoolConfig;
import com.mq.model.WorksPoolConfig;
import com.mq.redis.queue.RedisQueueListener;
import com.mq.redis.queue.executor.RedisQueueAckExecutor;
import com.mq.redis.queue.executor.RedisQueueWorkerExecutor;


/**
 * 处理中心核心类，考虑队列取出消息后将执行删除消息，采用策略：1. 使用redis自带复制消息，2.自定义传送至hash表(个人采用这个)
 * @author wenfei
 *
 * @param <T>
 * @param <V>
 */

@Controller
public class RedisQueue<T, V> implements InitializingBean, DisposableBean {

	// 初始化redis链接
	@Autowired
	private RedisTemplate redisTemplate;
	// 定义队列key
	private String key = "redisQueue_1";

	// 定义hashkey
	private String hashKey = "redisDoing";

	// 拿到当前key的二进制
	private byte[] rawKey; //
	// 获取连接工厂数据,
	private RedisConnectionFactory factory;
	// 获取连接属性,因RedisTemplate无法直接操作BRPOP/BLPOP
	private RedisConnection connection;
	// 初始化队列集合
	private BoundListOperations<String, T> listOperations;
	private BoundHashOperations<String, String, T> hashOperations;
	// 考虑互斥锁情况
	private Lock lock = new ReentrantLock();
	// 消费消息监听(线程主调)
	@Autowired
	private RedisQueueListener listener;
	@Autowired
	private WorksPoolConfig config; // 注入 配置
	@Autowired
	private AckPoolConfig ackPoolConfig;

	private RedisQueueWorkerExecutor queueWorkerExecutor;
	private RedisQueueAckExecutor queueAckExecutor;

	private Thread listenerThread;
	private boolean isClosed;


	@SuppressWarnings("all")
	@Override
	public void afterPropertiesSet() throws Exception {
		// 获取工厂
		factory = redisTemplate.getConnectionFactory();
		// 获取连接属性
		connection = RedisConnectionUtils.getConnection(factory);
		// 序列化key
		rawKey = redisTemplate.getKeySerializer().serialize(key);
		// 初始hash表
		hashOperations = redisTemplate.boundHashOps(hashKey);
		// 初始队列
		listOperations = redisTemplate.boundListOps(key);
		if (listener != null) {
			// 启动线程池
			System.out.println("[ RedisQueue Worker Executor Starting ] ....");
			queueWorkerExecutor = new RedisQueueWorkerExecutor(config);
			System.out.println("[ SUCCESS ] ....");

			// 启动ack调度线程池
			System.out.println("[ RedisQueue ACK Timer Starting ] ....");
			Timer timer = new Timer();
			Task task = new Task();
			timer.schedule(task, new Date(), 20000);
			System.out.println("[ SUCCESS ] ....");
			queueWorkerExecutor.execute(new ListenerThread());

		}
	}

	/**
	 * 从队列中获取首位消息(后进后出),命令：BLPOP timeout：0表示阻塞时间无限延长
	 */
	@SuppressWarnings("unchecked")
	public T takeBybLPop(int timeout) throws InterruptedException {
		// 允许被中断
		lock.lockInterruptibly();
		try {
			List<byte[]> results = connection.bLPop(timeout, rawKey);
			if (CollectionUtils.isEmpty(results)) {
				return null;
			}
			return (T) redisTemplate.getValueSerializer().deserialize(
					results.get(1));
		} finally {
			lock.unlock();
		}
	}

	/**
	 * 采用 从队列中获取首位消息(先进后出),命令：BRPOP timeout：0表示阻塞时间无限延长
	 */
	public T takeBybRPop(int timeout) throws InterruptedException {
		lock.lockInterruptibly();
		try {
			List<byte[]> results = connection.bRPop(timeout, rawKey);
			if (CollectionUtils.isEmpty(results)) {
				return null;
			}
			return (T) redisTemplate.getValueSerializer().deserialize(
					results.get(1));
		} finally {
			lock.unlock();
		}
	}

	class ListenerThread extends Thread {
		int i = 0;

		@Override
		public void run() {
			try {
				while (true) {
					System.out.println("[RedisQueue Read Startting]");
					if (i == 0) {
						//Thread.sleep(20000); // 测试读取顺序
					}
					i = 1;
					T value = takeBybRPop();// 将执行堵塞操作
					// 逐个执行
					if (value != null) {
						try {
							// 拿到时间戳
							long timeMills = System.currentTimeMillis();
							// 拿到线程名
							String threadName = Thread.currentThread()
									.getName();
							// 拼接成hashKey
							String hashOptKey = String.valueOf(timeMills) + "_"
									+ threadName;
							// 放入hash表
							hashOperations.put(hashOptKey, value);
							listener.onMessage(hashOptKey, value);
							// System.out.println("收到消息：" + value);
							// System.out.println("当前队列长度：" +
							// listOperations.size());

						} catch (Exception e) {
							//
						}
					}
				}
			} catch (InterruptedException e) {
				//
			}
		}
	}

	@SuppressWarnings("all")
	class Task extends TimerTask {
		@Override
		public void run() {
			
			System.err.println("[ RedisQueue ACK Timer Running...]");

			Map<String,T> dataMap =  hashOperations.entries();
			System.err.println("[ RedisQueue ACK Timer Running...]");
			if (dataMap != null) {
				for (Map.Entry<String, T> entry : dataMap.entrySet()) {
					// 处理业务推送
					try {
						
						String ackKey = entry.getKey();
								//(String) redisTemplate.getKeySerializer().deserialize((byte[])entry.getKey());
						String timeMills = ackKey.split("_")[0];
						String threadName = ackKey.split("_")[1];

						// 验证是否过期,需处理
						boolean ifOld = true;

						// 过期的情况,验证当前线程中是否存在当前线程任务
						if (ifOld) {
							// 需处理
							ThreadMXBean tmx = ManagementFactory.getThreadMXBean();
							// ThreadInfo info = tmx.
							// System.out.println(info.getThreadState());

						}
						new RedisQueueAckExecutor(config)
								.execute(new AckEntity(ackKey, threadName, ""));
					} catch (Exception e) {

						// 是否考虑计数(异常计数器)

					}
				}

			}

		}
	}

	class AckEntity extends Thread {

		private String id, name, type;
		
		
		@Autowired
		RedisQueue<String, V> redisQueue;

		public AckEntity(String id, String name, String type) {
			this.id = id;
			this.name = name;
			this.type = type;
		}

		@Override
		public void run() {

			// 处理发起, 获取考虑future,或catch捕获
			System.err.println("-------->>" + id + "," + name + "执行成功");
			boolean istrue = true;
			
			if(istrue){
				// 成功后
				removeFromHash(id);
			}else
				//失败了继续添加到队列中
				redisQueue.pushFromHead(id);

		}
	}

	public T takeFromHead() throws InterruptedException {
		return takeBybLPop(0);
	}

	public T takeBybRPop() throws InterruptedException {
		return takeBybRPop(0);
	}

	@Override
	public void destroy() throws Exception {
		if (isClosed) {
			return;
		}
		shutdown();
		RedisConnectionUtils.releaseConnection(connection, factory);
	}

	private void shutdown() {
		try {
			listenerThread.interrupt();
		} catch (Exception e) {
			//
		}
	}
	
	public T takeFromTail() throws InterruptedException {
		return takeBybLPop(0);
	}

	/**
	 * 从队列的头，插入
	 */
	public void pushFromHead(T value) {
		listOperations.leftPush(value);
	}

	public void pushFromTail(T value) {
		listOperations.rightPush(value);
	}

	/**
	 * @return null if no item in queue
	 */
	public T removeFromHead() {
		return listOperations.leftPop();
	}

	public T removeFromTail() {
		return listOperations.rightPop();
	}

	public void removeFromHash(String key) {
		hashOperations.delete(key);
	}
	public void setRedisTemplate(RedisTemplate redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	public void setListener(RedisQueueListener listener) {
		this.listener = listener;
	}

	public void setKey(String key) {
		this.key = key;
	}

}
